package com.ys.servers.common.mq;

import io.vertx.core.Future;
import io.vertx.core.Vertx;
import io.vertx.core.json.JsonObject;
import io.vertx.core.eventbus.DeliveryOptions;
import io.vertx.core.eventbus.EventBus;
import io.vertx.core.eventbus.MessageCodec;
import io.vertx.core.logging.Logger;
import io.vertx.core.logging.LoggerFactory;

import java.util.function.BiConsumer;
import java.util.function.Consumer;

/**
 * Created by taotao on 2017/2/23.
 */
public class EventBusImpl implements EventMQ {

    protected final Logger logger = LoggerFactory.getLogger(getClass());
    private EventBus eventBus;
    DeliveryOptions options;
    Vertx vertx;

    public EventBusImpl(Vertx vertx) {
        this.vertx = vertx;
        eventBus = vertx.eventBus();
        MessageCodec<InternalEventData, InternalEventData> myCodec = new EventBusCodec();
        try {
            eventBus.registerDefaultCodec(InternalEventData.class, myCodec);
        } catch (Exception e) {
            logger.info("register codec duplicate");
        }
        options = new DeliveryOptions().setCodecName(myCodec.name());
    }

    @Override
    public void send(String address, InternalEventData internalEventData, Consumer<InternalEventData> replyHandler) {
        eventBus.send(address, internalEventData, ar -> {
            if (ar.succeeded()) {
                InternalEventData respData = (InternalEventData) ar.result().body();
                replyHandler.accept(respData);
            } else {
                System.out.print("event bus send error:" + ar.cause());
            }
        });
    }

    @Override
    public void consumer(String address, BiConsumer<InternalEventData, Future<InternalEventData>> replyHandler) {
        eventBus.consumer(address, message -> {
            InternalEventData reqData = (InternalEventData) message.body();
            Future<InternalEventData> future = Future.future();
            future.setHandler(ar -> {
                if (ar.succeeded()) {
                    message.reply(ar.result(), options);
                } else {
                    logger.error(ar.cause());
                    InternalEventData data = new InternalEventData().setParas(new JsonObject().put("error", ar.cause()));
                    message.reply(data, options);
                }
            });
            replyHandler.accept(reqData, future);
        });
    }
}
